package com.alibaba.otter.node.etl.load.loader.mq.rabbit.strategy;

import com.alibaba.otter.node.etl.load.loader.mq.rabbit.MqUtils;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.exception.MqConsumeException;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.Collection;

@Slf4j
public class MqExceptionDispatcher {

    @Setter
    private FailureAction action;

    private BlockUntil endWithBlock = new BlockUntil();

    public void dispatch(MqConsumeException ex) {
        if (action == null) {
            return;
        }
        Class<? extends Throwable> exClass = ex.getOrigin().getClass();
        Collection<RescueAction> handlers = action.getExceptionHandler(exClass);
        int result = 0;
        for (RescueAction handler : handlers) {
            try {
                result = handler.handle(ex);
                if (actionSuccess(result) && breakChain(result)) {
                    break;
                }
            } catch (Exception e) {
                log.info("Failed to execute rescueAction: " + handler.getClass());
                result = -1;
                break;
            }
        }
        // 如果任何的RescueAction中出现异常，则直接跳出ActionChain，并Block消费线程，等待人工处理。
        if (result < 0) {
            log.error("RescueAction Failed, block until manually action. message: " + ex.getMsgCtx().getDeserializedMsg());
            endWithBlock.end(ex.getMsgCtx());
            return;
        }
        // 如果最后一个RescueAction执行成功，且允许跳过EndAction。则跳过EndAction
        if (actionSuccess(result) && skipEnd(result)) {
            return;
        }
        // 如果执行EndAction出错，则Block消费线程，等待人工处理。
        EndAction endAction = action.getEndStrategy(exClass);
        retry(endAction, ex);
    }

    private void retry(EndAction endAction, MqConsumeException ex) {
        Object message = ex.getMsgCtx().getDeserializedMsg();
        String endActionName = endAction.getClass().getName();
        try {
            endAction.end(ex.getMsgCtx());
            log.info("end action {} executed for message {}.", endActionName, message);
        } catch (Exception e) {
            log.warn("failed to execute endAction {} for message {}, retry in {} seconds.", endActionName, message, 1);
            MqUtils.sleep(1000);
            retry(endAction, ex);
        }
    }

    private boolean actionSuccess(int result) {
        return MqUtils.isBItNotZero(result, 0);
    }

    private boolean breakChain(int result) {
        return MqUtils.isBItNotZero(result, 4);
    }

    private boolean skipEnd(int result) {
        return MqUtils.isBItNotZero(result, 8);
    }
}
